import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object WaterMarkTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val stream1 = env.readTextFile("D:\\codes\\IDEA_prjects\\FlinkDemo\\data\\sensor.txt")
    val stream2 = stream1.map(
      data => {
        val strings = data.split(",")
        TempData(strings(0), strings(1).toLong, strings(2).toDouble)
      }
    )

    //TODO 单调递增数据
    stream2.assignAscendingTimestamps(_.time)
    //TODO  对于乱序数据
    stream2.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[TempData](Time.milliseconds(1000)) {
        override def extractTimestamp(element: TempData): Long = {
          element.time * 1000
        }
      }
    )


  }

  case class TempData(id: String, time: Long, temp: Double)
}
